import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel, RandomForestClassifier}
import org.apache.spark.ml.evaluation.{BinaryClassificationEvaluator, MulticlassClassificationEvaluator}
import org.apache.spark.ml.feature.{Binarizer, Bucketizer, ChiSqSelector, StandardScaler, VectorAssembler}
import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

import scala.collection.mutable.ListBuffer
object survied {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local").getOrCreate()
    import spark.implicits._

    val stream = spark.read.option("header", "true").option("inferSchema","true").csv("file:///D:\\IdeaProjects\\machine_learning\\test\\data\\survived.csv")
//    val stream = spark.read.option("header", "true").option("inferSchema","true").csv("/data/survived.csv")

    val source = (stream.drop("Ticket", "Name", "Cabin")
      .withColumn("Age", when(col("Age").isNull, (mean(col("Age").cast("int")).over(Window.orderBy())).cast("int")).otherwise(col("Age").cast("Int")))
      .withColumn("Sex", when(col("Sex") === "male", 1).otherwise(0))
      .na.drop()
      .withColumn("Embarked", dense_rank().over(Window.orderBy("Embarked")))
//      .withColumn("Pclass",col("Pclass").as("double"))
//      .withColumn("Sex",col("Sex").as("double"))
//      .withColumn("SibSp",col("SibSp").as("double"))
//      .withColumn("Parch",col("Parch").as("double"))
//      .withColumn("Emarked",col("Emarked").as("double"))
      )

    // 对年龄进行分桶
    val ageSplits = Array(Double.NegativeInfinity, 18, 35, 60, Double.PositiveInfinity)
    val AgeCast = new Bucketizer()
      .setInputCol("Age")
      .setOutputCol("AgeFeature")
      .setSplits(ageSplits).transform(source).drop("Age")

    // 对Fare进行极值化处理
    val FareCast = new Binarizer()
      .setInputCol("Fare")
      .setOutputCol("FareFeature")
      .setThreshold(100.0)
      .transform(AgeCast).drop("Fare")

    val resultDF = FareCast.withColumnRenamed("Survived", "label")

    val assembler = (new VectorAssembler()
      .setInputCols(
        resultDF.drop("PassengerID").drop("label").columns
      ).setOutputCol("features"))

    val StandData = new StandardScaler()
      .setInputCol("features")
      .setOutputCol("featuresScale")

    // 使用卡方检验方法做特征筛选
    val selector = new ChiSqSelector()
      .setFeaturesCol("featuresScale")
      .setLabelCol("label")
      .setOutputCol("featuresSelector")

    // 构建逻辑回归模型
    val logisticRegression = new LogisticRegression()
      .setLabelCol("label")
      .setFeaturesCol("featuresSelector")

    // 构建pipeline
    val pipeline = new Pipeline().setStages(Array(assembler, StandData, selector, logisticRegression))

    // 设置网络搜索最佳参数
    val paramMaps = new ParamGridBuilder()
      .addGrid(logisticRegression.regParam, Array(0.1, 0.01))
      .addGrid(selector.numTopFeatures, Array(5, 10, 15))
      .build()

    // 设置五折交叉验证
    val crossValidator = new CrossValidator()
      .setEstimator(pipeline)
      .setEvaluator(new BinaryClassificationEvaluator())  // 下方有详解
      .setEstimatorParamMaps(paramMaps)
      .setNumFolds(5)

    val Array(train,test) = resultDF.randomSplit(Array(0.8, 0.2))

    val model = crossValidator.fit(train)

    val prediction = model.bestModel.transform(test)

    // 通过模型得到最佳参数
    val bestModel = model.bestModel.asInstanceOf[PipelineModel]
    val lrModel = bestModel.stages(3).asInstanceOf[LogisticRegressionModel]
    println("正则化最佳参数: " + lrModel.getRegParam)
    println("卡方检验最佳特征数: " + lrModel.numFeatures)

    // 采用二分类评估器
    val evaluator = new BinaryClassificationEvaluator()
      .setLabelCol("label")
      .setRawPredictionCol("prediction")

    // AUC值，ROC曲线与坐标轴形成的面积，[0,1]
    println("AUC值: " + evaluator.setMetricName("areaUnderROC").evaluate(prediction))
    println("准确率: " + evaluator.setMetricName("areaUnderPR").evaluate(prediction))

    // setEvaluator()方法
    """
      |在setEvaluator方法中，您可以填写一个实现了Evaluator接口的评估器对象。评估器用于评估模型的性能，并根据指定的指标选择最佳模型。以下是一些可能的选项：
      |
      |BinaryClassificationEvaluator: 用于二分类问题的评估器，适用于计算ROC曲线下面积（AUC）等指标。
      |RegressionEvaluator: 用于回归问题的评估器，适用于计算均方根误差（RMSE）、平均绝对误差（MAE）等指标。
      |MulticlassClassificationEvaluator: 用于多分类问题的评估器，适用于计算准确率（accuracy）、F1分数等指标。
      |ClusteringEvaluator: 用于聚类问题的评估器，适用于计算聚类性能指标，如轮廓系数（silhouette coefficient）等。
      |""".stripMargin
    // 通过调参最后需选择了.setNumTrees(30).setMaxDepth(20)
//    val range = List("auto","all","onethird","sqrt","log2","n")
//    val buffer_test = new ListBuffer[Map[String, Double]]()
//    for (i <- range.indices){
//      val rfc = new RandomForestClassifier().setLabelCol("Survived").setSeed(1L).setNumTrees(30).setMaxDepth(20).setFeatureSubsetStrategy(range(i))
////      val rfc = new RandomForestClassifier().setLabelCol("Survived")
//
//      val model = rfc.fit(assembler.transform(train))
//
//      val predict_train = model.transform(assembler.transform(train))
//      val predict = model.transform(assembler.transform(test))
//
//      val accuracy = (new MulticlassClassificationEvaluator()
//        .setLabelCol("Survived")
//        .setMetricName("accuracy")
//        .evaluate(predict))
//
//      buffer_test.append(Map(range(i) -> accuracy))
//    }
//
//    buffer_test.sortBy(_.values).foreach(println)


//    val rfc = new RandomForestClassifier().setLabelCol("Survived").setSeed(1L).setNumTrees(30).setMaxDepth(20)
//
//    val model = rfc.fit(assembler.transform(train))
//
//    val predict_train = model.transform(assembler.transform(train))
//    val predict = model.transform(assembler.transform(test))
//
//    val accuracy_train = (new MulticlassClassificationEvaluator()
//      .setLabelCol("Survived")
//      .setMetricName("accuracy")
//      .evaluate(predict_train))
//
//    val accuracy = (new MulticlassClassificationEvaluator()
//      .setLabelCol("Survived")
//      .setMetricName("accuracy")
//      .evaluate(predict))
//
//    println(s"训练集精确度:$accuracy_train  测试集精确度:$accuracy")


  }
}
